#include "config.h"

#include "lib/libebc/ebc.h"
#include "lib/libbase/ecm_list.h"

static char	*server = "127.0.0.1";
static ecm_uint16	port;
static char	*mdname = "cbperf";
static ECMBOOL	server_mode;

static ecm_uint32	n_workers = 1;
static ecm_uint32	counts = 1;
static ecm_mutex_t	mutex;
static ecm_cond_t	cond_exit;
static ecm_cond_t	cond_delayed;
static ECMBOOL		going_to_shutdown;
static ECMBOOL		perf_started;
static ECMBOOL		delayed_cb;
static ecm_uint32	sleeptime;
static ecm_uint32	call_cnts;
static ecm_perf_t	perf;

typedef struct {
	ecm_uint32	key;
	ecm_uint64	tickcount;
	struct list_head	list;
} delayed_info_t;

static LIST_HEAD(delayed);

static void
usage(void)
{
	printf(
"Usage: cbperf <options>\n"
"    -s <server>: ebc server\n"
"    -p <port>: ebc port\n"
"    -m <modname>: ebc server modname\n"
"    -S: server mode\n"
"    -w <count>: client-side worker thread count\n"
"    -n <count>: calls per worker\n"
"    -t <msecs>: sleep time in call back(in msecs)\n"
"    -d: delayed callback\n"
		);
}

static void
parse_args(int argc, char *argv[])
{
	int	c;

	while ((c = ecm_getopt(argc, argv, "s:p:m:Sw:n:t:dh")) != -1) {
		switch (c) {
		case 's':
			server = ecm_strdup(ecm_optarg);
			break;
		case 'p':
			if (ecm_sscanf(ecm_optarg, "%hu", &port) != 1) {
				usage();
				exit(1);
			}
			break;
		case 'm':
			mdname = ecm_strdup(ecm_optarg);
			break;
		case 'S':
			server_mode = ECM_TRUE;
			break;
		case 'w':
			if (ecm_sscanf(ecm_optarg, "%u", &n_workers) != 1) {
				usage();
				exit(1);
			}
			break;
		case 'n':
			if (ecm_sscanf(ecm_optarg, "%u", &counts) != 1) {
				usage();
				exit(1);
			}
			break;
		case 't':
			if (ecm_sscanf(ecm_optarg, "%u", &sleeptime) != 1) {
				usage();
				exit(1);
			}
			break;
		case 'd':
			delayed_cb = ECM_TRUE;
			break;
		case 'h':
			usage();
			exit(0);
		default:
			usage();
			exit(1);
		}
	}
}

static void
print_perf(ecm_perf_t perf)
{
	ecm_show_perf(perf, ecm_stdout, "Throughput: %ccalls/sec, User:%u, Sys:%s\n");
}

static void
add_perf(void)
{
	if (!perf_started) {
		ecm_start_perf(perf);
		perf_started = ECM_TRUE;
	}
	call_cnts++;
	ecm_add_perf_data(perf, 0, 1);
}

static sr_t *
cb_test(ecm_ebc_t ebc, ebc_addr_t *saddr, ebc_addr_t *daddr, sr_t *sr, ecm_ptr_t ctx)
{
	if (delayed_cb) {
		delayed_info_t	*pinfo;

		ecm_mutex_lock(mutex);
		pinfo = ecm_malloc(sizeof(delayed_info_t));
		ebc_delay_msgcb(ebc, &pinfo->key);
		pinfo->tickcount = ecm_tickcount() + sleeptime;
		list_add_tail(&pinfo->list, &delayed);
		ecm_cond_signal(cond_delayed);
		ecm_mutex_unlock(mutex);
		return NULL;
	}
	if (sleeptime > 0)
		ecm_sleep(sleeptime);
	ecm_mutex_lock(mutex);
	add_perf();
	ecm_mutex_unlock(mutex);
	return sr_create_bool(ST_BOOL, ECM_TRUE);
}

static void
do_listen(ecm_ptr_t ctx)
{
	ebc_app_loop(ctx, 0);
}

static void
do_delayed(ecm_ptr_t ctx)
{
	ecm_ebc_t	ebc = (ecm_ebc_t)ctx;
	delayed_info_t	*pinfo;

	ecm_mutex_lock(mutex);
	while (!going_to_shutdown) {
		if (list_empty(&delayed))
			ecm_cond_wait(cond_delayed, mutex, NULL);
		else {
			ecm_uint64	curtime = ecm_tickcount();
			pinfo = list_entry(delayed.next, delayed_info_t, list);
			if (pinfo->tickcount > curtime) {
				ecm_uint32	timeout = pinfo->tickcount - curtime;
				ecm_cond_wait(cond_exit, mutex, &timeout);
				continue;
			}
			list_del(&pinfo->list);
			add_perf();
			ebc_reply_delayed_msgcb(ebc, pinfo->key, sr_create_bool(ST_BOOL, ECM_TRUE));
			ecm_free(pinfo);
		}
	}
	ecm_mutex_unlock(mutex);
}

static int
do_server(void)
{
	ecm_ebc_t	ebc;
	ecm_thread_t	listener, delay_thread;
	ecm_uint32	local_call_cnts = call_cnts;
	ebc_msgcb_flags_t	flags = 0;

	if (ebc_connect(server, port, mdname, 5000, &ebc) != EBC_OK) {
		ecm_fprintf(ecm_stderr, "cannot connect: %s:%hu\n", server, port);
		return 1;
	}
	perf = ecm_create_perf(1);
	if (delayed_cb)
		flags += EBC_MSGCB_DELAY;
	ebc_msgcb_add(ebc, "test", cb_test, NULL, flags);
	ecm_thread_create(&listener, do_listen, ebc);
	if (delayed_cb)
		ecm_thread_create(&delay_thread, do_delayed, ebc);

	ecm_mutex_lock(mutex);
	while (!going_to_shutdown) {
		ecm_uint32	timeout = 2000;

		if (call_cnts == local_call_cnts) {
			if (perf_started) {
				ecm_stop_perf(perf);
				perf_started = ECM_FALSE;
			}
		}
		else {
			print_perf(perf);
			local_call_cnts = call_cnts;
		}
		ecm_cond_wait(cond_exit, mutex, &timeout);
	}
	ecm_mutex_unlock(mutex);
	ecm_stop_perf(perf);
	ebc_exit_app_loop(ebc);
	ecm_thread_free(listener);
	if (delayed_cb)
		ecm_thread_free(delay_thread);
	ebc_close(ebc);

	return 0;
}

static void
do_work(ecm_ptr_t ctx)
{
	ecm_ebc_t	ebc;
	sr_t		*sr;
	ebc_addr_t	addr;
	ebc_err_t	err;
	ecm_uint32	i;

	if (ebc_connect(server, port, NULL, 5000, &ebc) != EBC_OK) {
		ecm_fprintf(ecm_stderr, "cannot connect: %s:%hu\n", server, port);
		return;
	}

	ebc_build_addr(&addr, ".", mdname, "test");
	for (i = 0; i < counts && !going_to_shutdown; i++) {
		sr = sr_create_str(ST_STR, "hello");
		err = ebc_dcall(ebc, &addr, &sr);
		sr_free(sr);
		if (err != EBC_OK) {
			ecm_fprintf(ecm_stderr, "call error: %d\n", err);
			return;
		}
	}
}

static int
do_client(void)
{
	ecm_thread_t	*threads;
	ecm_uint32	i;

	threads = (ecm_thread_t *)ecm_malloc(sizeof(ecm_thread_t) * n_workers);
	for (i = 0; i < n_workers; i++)
		ecm_thread_create(&threads[i], do_work, NULL);
	for (i = 0; i < n_workers; i++)
		ecm_thread_free(threads[i]);
	return 0;
}

static void
signal_handler(ecm_signo_t signo)
{
	switch (signo) {
	case ECM_SIGINT:
	case ECM_SIGTERM:
		going_to_shutdown = ECM_TRUE;
		ecm_mutex_lock(mutex);
		ecm_cond_broadcast(cond_exit);
		ecm_cond_broadcast(cond_delayed);
		ecm_mutex_unlock(mutex);
		break;
	default:
		break;
	}
}

int
main(int argc, char *argv[])
{
	ebc_initialize();

	ecm_signal(ECM_SIGINT, signal_handler);
	ecm_signal(ECM_SIGTERM, signal_handler);

	mutex = ecm_mutex_create(ECM_FALSE);
	cond_exit = ecm_cond_create();
	cond_delayed = ecm_cond_create();

	parse_args(argc, argv);

	if (server_mode)
		return do_server();
	else
		return do_client();
}
